package com.etc.wc.main;

import com.etc.wc.utils.PropertyHolderLazy;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.File;
import java.io.FilenameFilter;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.logging.Logger;

public class CoolectTask extends TimerTask {
    @Override
    public void run() {
        try {
            Properties pro = PropertyHolderLazy.getProps();
            //构造
            Logger logger = Logger.getLogger("logRollingFile");

            //获取本次采集时的日期
            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd-HH");
            String date = dateFormat.format(new Date());

            //获取本地文件路径
            File srcDir = new File(pro.getProperty("LOG_SOURCE_DIR"));
            //列出日志源目录中需要采集的文件
            File[] listFile = srcDir.listFiles(new FilenameFilter() {
                @Override
                public boolean accept(File dir, String name) {
                    if (name.startsWith(pro.getProperty("LOG_FILTER_NAME"))) {
                        return true;
                    }
                    return false;
                }
            });
            //记录日志
            logger.info("探测到如下文件需要采集" + Arrays.toString(listFile));

            //将要采集的文件移动到待上传临时目录
            File toUploadDir = new File(pro.getProperty("LOG_TOUPLOAD_DIR"));
            for (File file : listFile) {
                FileUtils.moveFileToDirectory(file, toUploadDir, true);
            }

            //记录日志
            logger.info("上述文件移动到了待上传目录中" + toUploadDir.getAbsolutePath());

            //构造一个HDFS的客户端对象

            FileSystem fileSystem = FileSystem.get(new URI(pro.getProperty("HDFS_URI")), new Configuration(), "root");
            File[] toUploadFiles = toUploadDir.listFiles();

            //检查HDFS中日期是否存在。如果不存在则创建
            Path hdfsPath = new Path("/log/" + date);
            if (!fileSystem.exists(hdfsPath)) {
                fileSystem.mkdirs(hdfsPath);
            }

            //检查本地的备份目录是否存在不存在  则创建
            File backupDir = new File(pro.getProperty("LOG_BACKUP_BASE_DIR") + date);
            if (!backupDir.exists()) {
                backupDir.mkdirs();
            }

            for (File f : toUploadFiles) {
                //HDFS目标路径
                Path destPath = new Path(hdfsPath + pro.getProperty("HDFS_FILE_QIAN") + UUID.randomUUID() + pro.getProperty("HDFS_FILE_HOU"));
                //传输文件到hdfs并改名字
                fileSystem.copyFromLocalFile(new Path(f.getAbsolutePath()), destPath);
                //记录日志
                logger.info("文件传输到HDFS完成" + f.getAbsolutePath() + "---->" + destPath);
                //将传输完成的文件移动到备忘目录
                FileUtils.moveFileToDirectory(f, backupDir, true);
                //记录日志
                logger.info("文件备份完成" + f.getAbsolutePath() + backupDir);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
